package com.atguigu.userprofile.task;

import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.constant.ConstCode;
import com.atguigu.userprofile.dao.TagInfoDao;
import com.atguigu.userprofile.util.ClickhouseUtil;
import com.atguigu.userprofile.util.MyPropsUtil;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;

import javax.swing.text.TabExpander;
import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;

/**
 * 将hive中计算好的标签宽表迁移到Clickhouse中
 *
 * 任务步骤:
 *
 * 1. 获取外部传入的参数 taskId 、 busiDate
 *
 * 2. 明确宽表的列， 通过被计算的标签来确定
 *
 * 3. 动态建表
 *
 * 4. 组织SQL:
 *    4.1 select(Hive) ....
 *    4.2 insert(Clickhouse)
 *
 */
public class TaskExport {

    public static void main(String[] args) {
        //1. 获取外部传入的参数 taskId 、 busiDate

        String taskId = args[0] ;
        String busiDate = args[1] ;

        //2. 明确宽表的列， 通过被计算的标签来确定
        List<TagInfo> tagInfos = TagInfoDao.selectTagInfosWithTaskStatusEnable();

        //3. 动态建表
        /*
          create table if not exists [ckDbName].[tableName]
          (
            uid String ,
            [tagColumns]
          )
          engine=MergeTree
          order by uid


          引擎选择:
            MergeTree :
            ReplacingMergeTree : 去重 , 一般考虑去重，主要是想做到幂等。
                                        一般在实时处理中需要重点考虑去重或者幂等。
                                         离线处理一般都简单粗暴，任务重跑直接删除上一次任务的结果。
            SummingMergeTree :   聚合

            Replcated...MergeTree : 副本 ， 数据高可用（高可用的代价比较高）  ， 一般业务服务都要高可用。

         */
        String upDbName = MyPropsUtil.get(ConstCode.UP_DBNAME) ;
        String ckDbName = MyPropsUtil.get(ConstCode.CK_DBNAME) ;
        String tableName = "up_merge_" + busiDate.replace("-" , "") ;

        //动态处理标签列
        String tagColumns = tagInfos.stream().map(
                tagInfo -> tagInfo.getTagCode().toLowerCase() + " String "
        ).collect(Collectors.joining(" , "));

        String createTable = " create table if not exists " + ckDbName+ "." + tableName +
                " (" +
                " uid String , " + tagColumns +
                " ) " +
                " engine=MergeTree " +
                " order by uid " ;

        System.out.println("createTable : " + createTable);

        String dropTable = " drop table if exists " + ckDbName + "." + tableName ;


        //删表
        ClickhouseUtil.executeSql(dropTable);

        //建表
        ClickhouseUtil.executeSql(createTable);


        //4.  组织SQL:
          //4.1 select(Hive) ....
        //动态处理标签列
        String selectColumns = tagInfos.stream().map(
                tagInfo -> tagInfo.getTagCode().toLowerCase()
        ).collect(Collectors.joining(" , "));
        String selectSql = " select uid , " + selectColumns + " from " + upDbName + "." + tableName ;
        SparkConf sparkConf = new SparkConf().setAppName("task_export_app");//.setMaster("local[*]");
        SparkSession sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate();
        Dataset<Row> dataset = sparkSession.sql(selectSql);

        //4.2 insert(Clickhouse)
        // 使用SparkSql支持的jdbc的操作将数据写到Clickhouse中
        String  ckUrl = MyPropsUtil.get(ConstCode.CK_URL) ;
        dataset.write()
                .mode(SaveMode.Append)
                .option("driver" , "ru.yandex.clickhouse.ClickHouseDriver")
                .option("numPartitions" , 4)  // 分区数， 决定并行读和写
                .option("batchsize" , 500) // 批次大小
                .option("isolationLevel" ,"NONE" ) // 事务隔离级别
                .jdbc(ckUrl , tableName , new Properties());
    }
}
